package com.atguigu.flink.datastramapi.transform;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by Smexy on 2023/2/25
 * 
 *  统计每种传感器的vc和
 */
public class Demo7_Process
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        
                env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   .keyBy(WaterSensor::getId)
                   //<K, I, O>
                   .process(new KeyedProcessFunction<String, WaterSensor, String>()
                   {
                       //后续会使用 状态API 解决
                       //private int sumVc ;
                       private Map<String,Integer> sumVc = new HashMap<>();
                       /*
                        来一条数据，调用一次处理。
                                Context ctx：  执行环境。 从中获取很多工具

                                Collector<String> out： 收集输出的工具
                        */
                       //属性

                       @Override
                       public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                           Integer oldVc = sumVc.getOrDefault(value.getId(), 0);
                           oldVc += value.getVc();
                           sumVc.put(value.getId(),oldVc);

                           //输出
                           out.collect(ctx.getCurrentKey()+":"+ oldVc);
                       }
                   })
                   .print();
        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
        
    }
}
